//package com.bs.sink;
//
//import com.alibaba.fastjson.JSONObject;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
//
//import java.sql.*;
//import java.util.Optional;
//
///**
// * @author bingshao
// * @date 2022/11/4
// **/
//public class MysqlSink extends RichSinkFunction<String> {
//
//    private Connection connection;
//    private PreparedStatement statementInsert, statementUpdateOnlyOrder, statementSelectUser, statementUpdateOnlyUser, statementSelectOrder;
//
//    @Override
//    public void open(Configuration parameters) throws Exception {
//        super.open(parameters);
//        Class.forName("com.mysql.cj.jdbc.Driver");
//        connection = DriverManager.getConnection("jdbc:mysql://110.42.233.197:3306/bingshao?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai", "root", "bs1354361838");
//        statementInsert = connection.prepareStatement("insert into target(id,name,product_name) values (?,?,?)");
////        statementSelect = connection.prepareStatement("select id ,name ,product_name from target where id = ?");
//        statementUpdateOnlyOrder = connection.prepareStatement("update target set product_name = ? where id = ?");
//        statementUpdateOnlyUser = connection.prepareStatement("update target set name = ? where id = ?");
//        statementSelectUser = connection.prepareStatement("select id ,name  from user where id = ?");
//        statementSelectOrder = connection.prepareStatement("select id ,user_id,product_name from `order` where user_id = ?");
//    }
//
//    @Override
//    public void invoke(String value, Context context) throws Exception {
//        JSONObject jsonObject = JSONObject.parseObject(value);
//        JSONObject after = JSONObject.parseObject(JSONObject.toJSONString(jsonObject.get("after")));
//        String type = jsonObject.get("type").toString();
//        if (Optional.ofNullable(jsonObject.get("tableName")).isPresent()) {
//            String tableName = jsonObject.get("tableName").toString();
//            // 监测到order说明order进行了操作，需要在target表进行改动
//            if ("order".equals(tableName)) {
//                // TODO 在此不做空值判断
//                long id = Long.parseLong(after.get("id").toString());
//                long userId = Long.parseLong(after.get("user_id").toString());
//                String productName = after.get("product_name").toString();
//                if ("create".equals(type)) {
//                    // 新增
//                    statementSelectUser.setLong(1, userId);
//                    statementSelectUser.execute();
//                    ResultSet resultSetOrder = statementSelectUser.getResultSet();
//                    while (resultSetOrder.next()) {
//                        String name = resultSetOrder.getString("name");
//                        statementInsert.setLong(1, id);
//                        statementInsert.setString(2, name);
//                        statementInsert.setString(3, productName);
//                        statementInsert.execute();
//                    }
//                }
//                if ("update".equals(type)) {
//                    // 修改
//                    statementUpdateOnlyOrder.setString(1, productName);
//                    statementUpdateOnlyOrder.setLong(2, id);
//                    statementUpdateOnlyOrder.execute();
//                }
//
//            }
//            if ("user".equals(tableName)) {
//                long id = Long.parseLong(after.get("id").toString());
//                String name = after.get("name").toString();
//                if ("update".equals(type)) {
//                    // 修改
//                    statementSelectOrder.setLong(1, id);
//                    statementSelectOrder.execute();
//                    ResultSet resultSet = statementSelectOrder.getResultSet();
//                    while (resultSet.next()) {
//                        Long targetId = resultSet.getLong("id");
//                        statementUpdateOnlyUser.setString(1, name);
//                        statementUpdateOnlyUser.setLong(2, targetId);
//                        statementUpdateOnlyUser.execute();
//                    }
//
//                }
//                if ("create".equals(type)) {
//                    // nothing to do
//                }
//            }
//        }
//
//    }
//
//    @Override
//    public void close() throws Exception {
//        super.close();
//        if (statementInsert != null) {
//            statementInsert.close();
//        }
////        if (statementSelect != null) {
////            statementSelect.close();
////        }
//        if (statementUpdateOnlyOrder != null) {
//            statementUpdateOnlyOrder.close();
//        }
//        if (statementSelectUser != null) {
//            statementSelectUser.close();
//        }
//        if (statementUpdateOnlyUser != null) {
//            statementUpdateOnlyUser.close();
//        }
//        if (statementSelectOrder != null) {
//            statementSelectOrder.close();
//        }
//        if (connection != null) {
//            connection.close();
//        }
//    }
//}
